package com.junmo.www.RxJava;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Scheduler;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.parallel.ParallelFlowable;
import io.reactivex.schedulers.Schedulers;
import org.testng.annotations.Test;

import java.util.ArrayList;
import java.util.List;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;



public class ParallelismTest {
    @Test
    public void test(){

    }
    @Test
    public void test05() throws InterruptedException {
        ParallelFlowable parallelFlowable = Flowable.range(1 , 100) .parallel() ;
        parallelFlowable.runOn(Schedulers.io ())
                .map(new Function<Integer, Object>() {
                    @Override
                    public Object apply(@NonNull Integer integer) throws Exception {
                        return integer.toString();
                    }
                })
                .sequential()
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String str) throws Exception {
                        System. out . println(str) ;
                    }
                });
        Thread.sleep(400000l);
    }

    @Test
    public void test04() throws InterruptedException {
        Executor executor = Executors.newFixedThreadPool(5);
        final Scheduler scheduler = Schedulers.from(executor);
        final AtomicInteger batch = new AtomicInteger(0);
        Observable.range(1,100)
                .groupBy(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer) throws Exception {
                        return batch.getAndIncrement()%5;
                    }
                })
                .flatMap(new Function<GroupedObservable<Integer, Integer>, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(@NonNull GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) throws Exception {
                        return integerIntegerGroupedObservable
                                .observeOn(scheduler)
                                .map(new Function<Integer, String>() {
                                    @Override
                                    public String apply(@NonNull Integer integer) throws Exception {
                                        System.out.println("-------------"+Thread.currentThread().getName());
                                        return integer.toString() ;
                                    }
                                });
                    }
                }).subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        System . out . println (o);
                    }
        });
        Thread.sleep(400000l);
    }


    @Test// flatMap 实现 并行编程
    public void test03() throws InterruptedException {
        int threadNum =Runtime.getRuntime().availableProcessors();
        Executor executor = Executors.newFixedThreadPool(threadNum);
        final Scheduler scheduler = Schedulers.from(executor);

        Observable.range(1,100).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                return Observable.just(integer)
                        .subscribeOn(scheduler)
                        .map(new Function<Integer, String>() {
                            @Override
                            public String apply(@NonNull Integer integer) throws Exception {
                                return integer.toString() ;
                            }
                        });
            }
        }).doFinally(new Action() {
            @Override
            public void run() throws Exception {
                scheduler.shutdown();
            }
        }).subscribe(new io.reactivex.functions.Consumer<String>() {
            @Override
            public void accept(String str) throws Exception {
                System .out.println(str);
            }
        });
        Thread.sleep(400000l);
    }


    @Test// flatMap 实现 并行编程
    public void test02() throws InterruptedException {
        Observable.range(1,100).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                return Observable.just(integer)
                        .subscribeOn(Schedulers.computation())
                        .map(new Function<Integer, String>() {
                            @Override
                            public String apply(@NonNull Integer integer) throws Exception {
                                return integer.toString() ;
                            }
                        });
            }
        }).subscribe(new io.reactivex.functions.Consumer<String>() {
            @Override
            public void accept(String str) throws Exception {
                System .out.println(str);
            }
        });
        Thread.sleep(400000l);
    }


    @Test// java8 并行编程
    public void test01(){
        List<Integer> list = new ArrayList<>();
        for(int i=0;i<100;i++) list.add(i);
        list.parallelStream().map(new java.util.function.Function<Integer, String>() {
            @Override
            public String apply(Integer integer) {
                return integer.toString() ;
            }
        }).forEach(new java.util.function.Consumer<String>() {
            @Override
            public void accept(String s) {
                System . out . println ("--  s ="+s +"; Current ThreadName="+Thread.currentThread(). getName() ) ;
            }
        });
    }
}
